#include "stream_server_source.h"
#include "media_stream.h"
#include "base_log.h"

namespace El {
namespace StreamServer {

StreamFramedSource::StreamFramedSource(UsageEnvironment &env, int32_t ch, int32_t type)
    : FramedSource(env), fCh(ch), fType(type)
{
    fHandle = Media::StreamSource::GetInstance(fCh, fType)
                  .Start(std::bind(&StreamFramedSource::onFrame, this, std::placeholders::_1));

    queue_ = std::unique_ptr<Base::BlockingQueue<Media::MediaFramePtr>>(
        new Base::BlockingQueue<Media::MediaFramePtr>("stream data", 10));
}

StreamFramedSource::~StreamFramedSource()
{
    queue_->SetActive(false);
    Media::StreamSource::GetInstance(fCh, fType).Stop(fHandle);

    queue_->Clear();
    queue_.reset();
}

void StreamFramedSource::deliverFrame0(void *clientData)
{
    ((StreamFramedSource *)clientData)->deliverFrame();
}

void StreamFramedSource::bufferExcessData(const uint8_t *data, size_t size)
{
    excessDataBuffer.insert(excessDataBuffer.end(), data, data + size);
}

void StreamFramedSource::deliverFrame()
{
    size_t totalDataSize = excessDataBuffer.size();
    Media::MediaFramePtr frame = queue_->Pop(0);
    if (frame) {
        totalDataSize += frame->GetLength();
        currentFrame_ = frame;
    }

    if (totalDataSize > 0) {
        if (totalDataSize > fMaxSize) {
            fFrameSize = fMaxSize;
            fNumTruncatedBytes = totalDataSize - fMaxSize;
        } else {
            fFrameSize = totalDataSize;
            fNumTruncatedBytes = 0;
        }

        size_t copiedSize = 0;

        // First, copy data from excessDataBuffer
        if (!excessDataBuffer.empty()) {
            size_t sizeToCopy = std::min(excessDataBuffer.size(), static_cast<size_t>(fMaxSize));
            memcpy(fTo, excessDataBuffer.data(), sizeToCopy);
            copiedSize += sizeToCopy;

            // Remove copied data from excessDataBuffer
            excessDataBuffer.erase(excessDataBuffer.begin(), excessDataBuffer.begin() + sizeToCopy);
        }

        // Then, copy data from the new frame if available
        if (frame && copiedSize < fMaxSize) {
            size_t remainingSpace = fMaxSize - copiedSize;
            size_t sizeToCopy = std::min(static_cast<size_t>(frame->GetLength()), remainingSpace);
            memcpy(fTo + copiedSize, frame->GetBuffer(), sizeToCopy);
            copiedSize += sizeToCopy;

            // Buffer any remaining data from the new frame
            if (sizeToCopy < (size_t)frame->GetLength()) {
                bufferExcessData(frame->GetBuffer() + sizeToCopy, frame->GetLength() - sizeToCopy);
            }
        }

        fPresentationTime.tv_sec = currentFrame_->GetPts() / 1000000;
        fPresentationTime.tv_usec = currentFrame_->GetPts() % 1000000;

        // 以25帧每秒计算
        fDurationInMicroseconds = 40000; // 40 ms

        // LOG_INFO("StreamFramedSource: Delivered {} bytes (truncated {} bytes)", fFrameSize, fNumTruncatedBytes);
        FramedSource::afterGetting(this);
    } else {
        // No data available, schedule next delivery attempt
        nextTask() = envir().taskScheduler().scheduleDelayedTask(10000, // 10 ms
                                                                 (TaskFunc *)deliverFrame0, this);
    }
}

void StreamFramedSource::doGetNextFrame()
{
    deliverFrame();
}

void StreamFramedSource::onFrame(const Media::MediaFramePtr &frame)
{
    LOG_DEBUG("Received frame of size {} type {}", frame->GetLength(), frame->GetFrameType());
    if (frame->GetFrameType() == MEDIA_FRAME_I || frame->GetFrameType() == MEDIA_FRAME_P) {
        queue_->Push(frame);
    }
}

StreamServerMediaSubsession *StreamServerMediaSubsession::createNew(UsageEnvironment &env,
                                                                    int32_t ch,
                                                                    int32_t type,
                                                                    bool reuseFirstSource)
{
    return new StreamServerMediaSubsession(env, ch, type, reuseFirstSource);
}

StreamServerMediaSubsession::StreamServerMediaSubsession(UsageEnvironment &env,
                                                         int32_t ch,
                                                         int32_t type,
                                                         bool reuseFirstSource)
    : OnDemandServerMediaSubsession(env, reuseFirstSource), fCh(ch), fType(type)
{
}

FramedSource *StreamServerMediaSubsession::createNewStreamSource(unsigned clientSessionId, unsigned &estBitrate)
{
    (void)clientSessionId;
    estBitrate = 500; // kbps, modify as needed
    return H264VideoStreamFramer::createNew(envir(), new StreamFramedSource(envir(), fCh, fType));
}

RTPSink *StreamServerMediaSubsession::createNewRTPSink(Groupsock *rtpGroupsock,
                                                       unsigned char rtpPayloadTypeIfDynamic,
                                                       FramedSource *inputSource)
{
    (void)rtpPayloadTypeIfDynamic;
    (void)inputSource;
    return H264VideoRTPSink::createNew(envir(), rtpGroupsock, rtpPayloadTypeIfDynamic);
}

} // namespace StreamServer
} // namespace El